package cn.itcast.d_state;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;

//需求: 使用MapState对本地集合进行求和 (根据第一个字段分组)
public class MapStateDemo {
    public static void main(String[] args) throws Exception {
        /**
         * 1.获取流处理执行环境
         * 2.加载本地集合
         * 3.数据分组
         * 4.数据求和
         * 5.数据打印
         * 6.触发执行
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Tuple2<Long, Long>> source = env.fromCollection(Arrays.asList(
                Tuple2.of(1L, 4L),
                Tuple2.of(2L, 3L),
                Tuple2.of(3L, 1L),
                Tuple2.of(1L, 2L),
                Tuple2.of(3L, 2L),
                Tuple2.of(1L, 2L),
                Tuple2.of(2L, 2L),
                Tuple2.of(2L, 9L)));

        //1,8
        //2,14
        //3,3
        //3.数据分组
        source.keyBy(new KeySelector<Tuple2<Long, Long>, Long>() {
            @Override
            public Long getKey(Tuple2<Long, Long> value) throws Exception {
                return value.f0;
            }
        }).map(new RichMapFunction<Tuple2<Long, Long>, Object>() {
            MapState<Long, Long> ms = null;
            @Override
            public void open(Configuration parameters) throws Exception {

                //设置状态类型
                MapStateDescriptor<Long, Long> mapDesc = new MapStateDescriptor<>("mapDesc", TypeInformation.of(Long.class), TypeInformation.of(Long.class));
                ms = getRuntimeContext().getMapState(mapDesc);
            }

            @Override
            public Object map(Tuple2<Long, Long> value) throws Exception {

                //获取状态数
                Long valueLast = ms.get(value.f0);
                valueLast = valueLast == null?0l:valueLast;

                Tuple2<Long, Long> of = Tuple2.of(value.f0, value.f1 + valueLast);
                //更新状态
                ms.put(of.f0,of.f1);
                return of;
            }
        }).print();


        env.execute();
    }
}
